package com.example.demo.mqserver.datacenter;

import com.example.demo.common.BinaryTool;
import com.example.demo.common.MqException;
import com.example.demo.mqserver.core.MSGQueue;
import com.example.demo.mqserver.core.Message;
import lombok.extern.slf4j.Slf4j;

import java.io.*;
import java.util.LinkedList;
import java.util.Scanner;

/**
 * Created with IntelliJ IDEA.
 * Description:
 * User: 张
 * Date: 2024-07-30
 * Time: 0:29
 */
//针对存储消息进行管理
@Slf4j
public class MessageFileManager {
    //定义一个内部来表示该队列的统计信息(消息总数 有效消息数)
    //静态内部类(解耦合)
    static public class Stat {
        public int totalCount;//总消息数量
        public int validCount;//有效消息数量
    }

    // 约定消息文件所在的目录 和 文件名
    //这个方法是用来获取到指定队列对应的消息文件所在的路劲
    private String getQueueDir(String queueName) {
        return "./data/" + queueName;
        //约定每个消息队列的文件所在地址为 ./data/ + 队列名
    }

    //这个方法用来获取该队列的消息数据文件路径
    //注: 这是一个二进制文件
    private String getQueueDataPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data.txt";
    }

    //这个方法用来获取该队列的消息统计文件路径
    private String getQueueStatPath(String queueName) {
        return getQueueDir(queueName) + "/queue_stat.txt";
    }

    private Stat readStat(String queueName) {
        Stat stat = new Stat();
        //由于当前的消息统计文件是文本文件,可以直接使用Scanner 来读取文件内容
        try (InputStream inputStream = new FileInputStream(getQueueStatPath(queueName))) {
            Scanner scanner = new Scanner(inputStream);
            stat.totalCount = scanner.nextInt();//总消息数量
            stat.validCount = scanner.nextInt();//有效消息数量
        } catch (IOException e) {
            e.printStackTrace();
        }
        return stat;
    }

    private void writeStat(String queueName, Stat stat){
        //使用 PrintWrite 来写文件
        //OutputStream 打开文件的时候,默认情况下会把文件清空,此时相当于新的数据覆盖旧的数据  加多一个true参数的时候就是追加写入的意思(append = true)
        try(OutputStream outputStream = new FileOutputStream(getQueueStatPath(queueName))){
            PrintWriter printWriter = new PrintWriter(outputStream);
            printWriter.write(stat.totalCount + "\t" + stat.validCount);
            printWriter.flush();//刷新
        } catch (IOException e) {
            e.printStackTrace();
        }
    }


    //初始化 --- 以备后续扩展
    public void init() {
        //先不写入任何方法
    }


    //创建队列对应的文件目录

    /**
     * 此处出现过的问题 创建文件夹和文件使用的不是一个方法 一个是mkdirs() 一个是createNewFile
     * 不要在出现相同的错误
     * @param queueName
     * @throws IOException
     */
    public void createQueueFiles(String queueName) throws IOException {
        //1.先创建队列对应的消息目录
        File baseDir = new File(getQueueDir(queueName));
        if(!baseDir.exists()) {
            //不存在就创建这个目录
            boolean ok = baseDir.mkdirs();
            if (!ok) {
                throw new IOException("创建目录失败!! baseDir = " + baseDir.getAbsolutePath());
            }
        }
        //2. 创建队列数据文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()) {
            boolean ok = queueDataFile.createNewFile();
            if(!ok) {
                throw new IOException("创建文件失败!! queueDataFile = " + queueDataFile.getAbsolutePath());
            }
        }
        //3. 创建消息统计文件
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()) {
            boolean ok = queueStatFile.createNewFile();
            if (!ok) {
                throw new IOException("创建文件失败!! queueStatFile = " + queueStatFile.getAbsolutePath());
            }
        }
        // 4.给消息统计文件设定初始值 0\t0
        Stat stat = new Stat();
        stat.validCount = 0;
        stat.totalCount = 0;
        writeStat(queueName,stat);
    }

    //删除队列的目录和文件
    //队列是可以被删除的,当队列删除之后,对应的消息文件自然也要随之删除
    public void destroyQueueFiles(String queueName) throws IOException {
        //先删除文件
        File queueDataFile = new File(getQueueDataPath(queueName));
        boolean ok1 = queueDataFile.delete();
        File queueStatFile = new File(getQueueStatPath(queueName));
        boolean ok2 = queueStatFile.delete();
        File baseDir = new File(getQueueDir(queueName));
        boolean ok3 = baseDir.delete();
        if(!ok1 || !ok2 || !ok3) {
            //有任意一个删除失败都是都算整体删除失败
            throw new IOException("删除队列目录和文件失败 baseDir = " + baseDir.getAbsolutePath());
        }
    }

    //检查队列的目录和文件是否存在(检查方法)
    //生产者给BrokerServer生产消息了,就得去检查一下队列是否完整
    public boolean checkFileExits(String queueName) {
        // 队列的数据文件和统计文件是否都存在
        File queueDataFile = new File(getQueueDataPath(queueName));
        if(!queueDataFile.exists()) {
            log.info("队列数据文件不存在 queueDataFile = " + queueDataFile.getAbsolutePath());
            return false;
        }
        File queueStatFile = new File(getQueueStatPath(queueName));
        if(!queueStatFile.exists()) {
            log.info("队列统计文件不存在 queueStatFile = " + queueDataFile.getAbsolutePath());
            return false;
        }
        return true;
    }

    //将一个新的消息写入对应的文件文件 (线程安全) -- 上锁(针对队列进行加锁)
    public void sendMessage(MSGQueue queue, Message message) throws MqException, IOException {
        //1.检查一下当前要写入为文件是否存在
        if(!checkFileExits(queue.getName())) {
            throw new MqException("[MessageFileManager] 队列对应文件不存在!! queueName = " + queue.getName());
        }
        // 2.把Message对线进行序列化
        byte[] messageBinary = BinaryTool.toBytes(message);
        synchronized (queue) {
            // 3.先获取当前队列数据文件的长度,用这个来计算出该Message对象的offsetBeg 和 offsetEnd
            // 约定:新的消息进来的时候是默认插入到队尾,且头4个字节用来表示消息的长度
            // offsetBeg = 文件长度 +4  offsetEnd = 文件长度 + 4 + 新的消息长度  这两个数据是保存在内存中的
            File queueDataFile = new File(getQueueDataPath(queue.getName()));
            message.setOffsetBeg(queueDataFile.length() + 4);
            message.setOffsetEnd(queueDataFile.length() + 4 + messageBinary.length);
            //4.写入消息到数据文件,追加写入
            try(OutputStream outputStream = new FileOutputStream(queueDataFile,true)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    //先写消息的长度,占4个直接
                    dataOutputStream.writeInt(messageBinary.length);
                    //写入消息本体
                    dataOutputStream.write(messageBinary);
                }
            }
            //5.更细消息统计文件
            Stat stat = readStat(queue.getName());
            synchronized (stat) {
                stat.totalCount += 1;
                stat.validCount += 1;
                writeStat(queue.getName(),stat);
            }
        }
    }

    //删除消息
    //逻辑删除
    //1.先把文件中的这一段数据读出来,还原成Message对象 ---随机访问
    //2,把isValid 改成0
    //3,把上述数据重新写入文件
    public void deleteMessage(MSGQueue queue,Message message) throws IOException, ClassNotFoundException {
        //确保线程安全
        synchronized (queue) {
            try(RandomAccessFile randomAccessFile = new RandomAccessFile(getQueueDataPath(queue.getName()),"rw")) {
                // 1.先从文件中读取对应的Message数据
                byte[] bufferSrc = new byte[(int)(message.getOffsetEnd() - message.getOffsetBeg())];
                randomAccessFile.seek(message.getOffsetBeg());//将光标指向数据的开头
                randomAccessFile.read(bufferSrc);
                // 2.把读出来的二进制数据转换成Message对象
                Message diskMessage = (Message) BinaryTool.fromBytes(bufferSrc);
                // 3.修改IsValid数据
                // 此处无需修改内存中的Message对象的IsValid这个属性,IsValid只是用来辅助文件操作的,
                // 内存中要进行删除丢掉就行了
                diskMessage.setIsValid((byte) 0X0);
                // 4.重新写入文件
                byte[] bufferDest = BinaryTool.toBytes(diskMessage);
                randomAccessFile.seek(message.getOffsetBeg());//前面光标移动了,需要重新定位
                randomAccessFile.write(bufferDest);//数据覆盖
            }
            //更新统计文件
            Stat stat = readStat(queue.getName());
            if(stat.validCount > 0) {
                stat.totalCount -=1;
            }
            writeStat(queue.getName(),stat);
        }
    }


    // 读取所有消息内容,加载到内存中
    // 这个方法,准备在程序启动的时候,进行调用  服务器重启之后填补内存中的数据 ---- 无需加锁
    // 此处使用LikedList 主要是为了后续的头删操作(队列) 这里的参数只需要队列的名称,传个name过来就可以了
    public LinkedList<Message> loadAllMessageFromQueue(String queueName) throws IOException, MqException, ClassNotFoundException {
        LinkedList<Message> messages = new LinkedList<>();
        try(InputStream inputStream = new FileInputStream(getQueueDataPath(queueName))) {
            try(DataInputStream dataInputStream = new DataInputStream(inputStream)) {
                //使用这个变量来记录光标所在位置
                long currentOffset = 0;
                //循环读取
                while (true) {
                    // 1.读取文件长度 此处的readInt可能会读到文件的末尾(EOF)  readInt会抛出EOFException异常
                    int messageSize = dataInputStream.readInt();//预期消息长度
                    // 2.按照消息长度读取消息内容
                    byte[] buffer = new byte[messageSize];
                    int actualSize = dataInputStream.read(buffer);//实际消息长度
                    if(messageSize != actualSize) {
                        // 文件有问题(文件剩余空间不够)
                        throw new MqException("[MessageFileManager] 文件格式错误! queueName = " + queueName);
                    }
                    // 3.数据转换
                    Message message = (Message) BinaryTool.fromBytes(buffer);
                    // 4.判断该消息是否有效
                    if(message.getIsValid() != 0X1) {
                        //无效数据也需要更新数据
                        currentOffset += (4 + messageSize);
                        continue;
                    }
                    // 5.添加有效数据
                    // 在进行计算OffsetBeg 和 OffsetEnd 的时候需要一个计数器
                    // DataInputStream类 没有直接提供光标所在位置的方法
                    message.setOffsetBeg(currentOffset + 4);
                    message.setOffsetEnd(currentOffset + 4 + messageSize);
                    currentOffset += (4 + messageSize);
                    messages.add(message);
                }
            } catch (EOFException e) {
                //这个 catch 是用来刹车的
                System.out.println("[MessageFileManager] 恢复数据完成!  queueName = " + queueName);
            }
        }
        return messages;
    }

    //判定当前队列是否要触发垃圾回收操作
    public boolean checkGC(String queueName) {
        //从消息统计文件读出对应数据
        Stat stat = readStat(queueName);
        if(stat.totalCount >= 2000 && (double)stat.validCount / (double) stat.totalCount < 0.5) {
            return true;
        }
        return false;
    }

    //约定触发GC之后新的文件的文件名称  ---  此处的文件只是负责暂时存储,等GC操作完成之后存放数据的文件名还是queue_data.txt
    private String getQueueDataNewPath(String queueName) {
        return getQueueDir(queueName) + "/queue_data_new.txt";
    }

    // 垃圾回收   ----   "未言胜,先言败"
    // 使用复制算法
    // 先创建一个新的文件 queue_data_new.txt 将旧的文件中的有效的消息复制一份在新的文件上
    // 最后删除旧的文件,将queue_data_new.txt 的文件名换回 queue_data.txt即可
    // 同时要记得更新消息统计文件
    public void gc (MSGQueue queue) throws MqException, IOException, ClassNotFoundException {
        // 在进行GC的时候,是针对消息文件进行大洗牌,要进行加锁
        synchronized (queue) {
            // 由于 GC 操作可能比较耗时,此处统计一下执行GC消耗的时间
            long gcBeg = System.currentTimeMillis();
            //1.创建新的文件
            File queueDataNewFile = new File(getQueueDataNewPath(queue.getName()));
            if(queueDataNewFile.exists()) {
                //正常情况下这个文件不应该存在的,如果存在就说明上次GC操作没有执行完毕
                throw new MqException("[MessageFileManager] gc 时,发现该队列的queue_Data_New 已经存在, queueName = " + queue.getName());
            }
            boolean ok = queueDataNewFile.createNewFile();//不存在就创建出来
            if(!ok) {
                throw new MqException("[MessageFileManager] 创建文件失败! queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }

            //2,从旧的文件中读取所有有效消息的对象
            LinkedList<Message> messages = loadAllMessageFromQueue(queue.getName());

            //3. 将有效消息读入文件中,msgQueues中已经全是有效的消息了
            try(OutputStream outputStream = new FileOutputStream(queueDataNewFile)) {
                try(DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                    for (Message message : messages) {
                        byte[] buffer = BinaryTool.toBytes(message);
                        //先写4个字节(消息的长度)
                        //message中的数据也要进行更新
                        message.setOffsetBeg(queueDataNewFile.length() + 4);
                        message.setOffsetEnd(queueDataNewFile.length() + 4 + buffer.length);
                        dataOutputStream.writeInt(buffer.length);
                        dataOutputStream.write(buffer);
                    }
                }
            }
            //4.删除旧的数据文件,并且把新的数据文件进行重命名
            File queueDataOldFile = new File(getQueueDataPath(queue.getName()));
            ok = queueDataOldFile.delete();
            if(!ok) {
                throw new MqException("[MessageFileManager] 删除旧的数据文件失败,queueDataOldFile = " + queueDataOldFile.getAbsolutePath());
            }
            //文件重命名  queue_data_new.txt -> queue_data.txt
            ok = queueDataNewFile.renameTo(queueDataOldFile);
            if(!ok) {
                throw new MqException("[MessageFileManager] 文件重命名失败,queueDataNewFile = " + queueDataNewFile.getAbsolutePath());
            }
            //5.更新统计文件
            Stat stat = readStat(queue.getName());
            stat.totalCount = messages.size();
            stat.validCount = messages.size();
            writeStat(queue.getName(),stat);

            long gcEnd = System.currentTimeMillis();
            System.out.println("[MessageFileManager] gc 执行完毕, queueName = " + queue.getName() + ", time = " +(gcEnd - gcBeg) + "ms" );
        }
    }

}